package com.hmall.common.utils;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@RequiredArgsConstructor
@Slf4j
public class RabbitMqHelper {

    private final RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange, String routingKey, Object msg) {
        rabbitTemplate.convertAndSend(exchange, routingKey, msg);
    }

    public void sendDelayMessage(String exchange, String routingKey, Object msg, int delay) {
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //添加延迟消息属性
                message.getMessageProperties().setDelay(delay);
                return message;
            }
        });
    }

    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public void sendMessageWithConfirm(String exchange, String routingKey, Object msg, int maxRetries) {
        String messageId = UUID.randomUUID().toString();
        CorrelationData cd = new CorrelationData(messageId);

        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("发送消息失败（未收到确认），ID: {}, 剩余重试次数: {}", messageId, maxRetries, ex);
                retryLater();
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                if (result.isAck()) {
                    log.info("消息发送成功（ACK），ID: {}", messageId);
                } else {
                    log.error("消息发送失败（NACK），ID: {}, 原因: {}，剩余重试次数: {}",
                            messageId, result.getReason(), maxRetries);
                    retryLater();
                }
            }

            private void retryLater() {
                if (maxRetries > 0) {
                    scheduler.schedule(() ->
                                    sendMessageWithConfirm(exchange, routingKey, msg, maxRetries - 1),
                            2, TimeUnit.SECONDS);
                } else {
                    log.error("消息重试次数耗尽，最终失败。消息 ID: {}, 内容: {}", messageId, msg);
                }
            }
        });

        try {
            rabbitTemplate.convertAndSend(exchange, routingKey, msg, cd);
        } catch (Exception e) {
            log.error("发送消息过程中异常，ID: {}, 剩余重试次数: {}", messageId, maxRetries, e);
            if (maxRetries > 0) {
                scheduler.schedule(() ->
                                sendMessageWithConfirm(exchange, routingKey, msg, maxRetries - 1),
                        2, TimeUnit.SECONDS);
            }
        }
    }

}